package SparkExercises.kejian5.LianXi2;

import com.google.gson.Gson;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.List;

public class LianXi2 {
    public static void main(String[] args) {
        /**
         *给定三份数据：其中violation.txt是违章条目对照表（违章代码、扣分数、罚款金额、违法内容），
         * records.txt是违章记录（日期 设备编号 车牌号 违章代码），
         * owner.txt是车主信息（车牌号、车主姓名、车主电话）
         * ，这三个文件列于列之间都是用tab分割。
         * 请写一个spark程序，
         * 查找出累计扣12分以上车辆信息（包含：车牌号、车主姓名、车主电话、总扣分），并以json格式保存到hdfs分布式文件系统中
         */

        SparkConf sparkConf = new SparkConf().setAppName("LianXi2").setMaster(args[0]);

        // 读数据
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
        JavaRDD<String> violationJavaRDD = javaSparkContext.textFile("hdfs://192.168.84.128:9000/data/violation.txt");
        JavaRDD<String> recordsJavaRDD = javaSparkContext.textFile("hdfs://192.168.84.128:9000/data/records(3).txt");
        JavaRDD<String> ownerJavaRDD = javaSparkContext.textFile("hdfs://192.168.84.128:9000/data/owner.txt");
//        JavaRDD<String> violationJavaRDD = javaSparkContext.textFile("./data/violation.txt");
//        JavaRDD<String> recordsJavaRDD = javaSparkContext.textFile("./data/records(3).txt");
//        JavaRDD<String> ownerJavaRDD = javaSparkContext.textFile("./data/owner.txt");

        // 创建键值对 violationJavaPairRDD
        JavaPairRDD<String, String> violationJavaPairRDD = violationJavaRDD.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                String[] strings = s.split("\t");  // s:"X01	2	200	超速10%-20%"
                return new Tuple2<>(strings[0], s);
            }
        });

        // 创建键值对 recordsJavaPairRDD
        JavaPairRDD<String, String> recordsJavaPairRDD = recordsJavaRDD.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                String[] string = s.split("\t");  // s:"2020-1-05	A301	CZ8463	X04"

                return new Tuple2<>(string[3], s);
            }
        });

        // 内连接 violation Join Records
        JavaPairRDD<String, Tuple2<String, String>> violationJoinRecords = violationJavaPairRDD.join(recordsJavaPairRDD);


        // 转换 ("车牌号","扣分数")
        JavaRDD<Tuple2<String, Integer>> map = violationJoinRecords.map(new Function<Tuple2<String, Tuple2<String, String>>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Tuple2<String, String>> s) throws Exception {
//                s:"(X06,(X06	3	150	逆向行驶,2020-1-18	A301	CZ8463	X06))"

                String key = s._2._2.split("\t")[2];  // s._2._2 = "2020-1-18	A301	CZ8463	X06"

//               s._2._1 = "X06	3	150	逆向行驶"
                Integer value = Integer.parseInt(s._2._1.split("\t")[1]);
                return new Tuple2<>(key, value);
            }
        });

        // 创建键值对 stringIntegerJavaPairRDD
        JavaPairRDD<String, Integer> stringIntegerJavaPairRDD = map.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2;
            }
        });

        JavaPairRDD<String, Integer> JavaPairRDD1 = stringIntegerJavaPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });

        JavaPairRDD<String, String> ownerJavaPairRDD = ownerJavaRDD.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                String s1 = s.split("\t")[0];
                return new Tuple2<>(s1, s);
            }
        });

        JavaPairRDD<String, Tuple2<String, Integer>> ownerJavaPairRDDJoinJavaPairRDD1 = ownerJavaPairRDD.join(JavaPairRDD1);

        JavaRDD<Tuple2<String, Integer>> values = ownerJavaPairRDDJoinJavaPairRDD1.values();


        // 转换为Cat对象
        JavaRDD<Cat> mapCat = values.map(new Function<Tuple2<String, Integer>, Cat>() {
            @Override
            public Cat call(Tuple2<String, Integer> T) throws Exception {
                String[] split = T._1.split("\t");

                return new Cat(split[0], split[1], split[2], T._2);
            }
        });
        mapCat.foreach(new VoidFunction<Cat>() {
            @Override
            public void call(Cat cat) throws Exception {
                System.out.println(cat);
            }
        });

//        String path = "hdfs://192.168.84.128:9000/out/out_b";
        String path = args[1];

        // 写数据到指定路径
        mapCat.saveAsTextFile(path);

    }
}

class Cat{
    private String license_plate_number;
    private String name;
    private String phone_number;
    private Integer sum;

    @Override
    public String toString() {
        return "Cat{" +
                "license_plate_number='" + license_plate_number + '\'' +
                ", name='" + name + '\'' +
                ", phone_number='" + phone_number + '\'' +
                ", sum=" + sum +
                '}';
    }

    public Cat(String license_plate_number, String name, String phone_number, Integer sum) {
        this.license_plate_number = license_plate_number;
        this.name = name;
        this.phone_number = phone_number;
        this.sum = sum;
    }

    public String getLicense_plate_number() {
        return license_plate_number;
    }

    public void setLicense_plate_number(String license_plate_number) {
        this.license_plate_number = license_plate_number;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getPhone_number() {
        return phone_number;
    }

    public void setPhone_number(String phone_number) {
        this.phone_number = phone_number;
    }

    public Integer getSum() {
        return sum;
    }

    public void setSum(Integer sum) {
        this.sum = sum;
    }
}